package com.zhuchang.self_rpc.test.common;

import com.zhuchang.self_rpc.test.codec.RpcRequest;
import com.zhuchang.self_rpc.test.codec.RpcResponse;
import jdk.nashorn.internal.ir.CallNode;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author RenPu
 * @Date 2021/8/18 16:42
 * @Version 1.0
 * @Description: future模型对象
 **/
@Slf4j
public class RpcFuture implements Future<Object> {

    //请求对象
    private RpcRequest rpcRequest;

    //响应对象
    private RpcResponse rpcResponse;

    //设置开始时间
    private long startTime;

    //等待回调的集合
    private List<RpcCallBack> pendingCallbacks = new ArrayList<RpcCallBack>();

    //创建可重入锁
    private Lock lock = new ReentrantLock();


    //创建异步线程池
    ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(65536));

    private static final long TIME_THRESHOLD = 5000;

    //基于AQS的自定义锁
    private Sync sync = new Sync();

    public RpcFuture(RpcRequest rpcRequest) {
        this.rpcRequest = rpcRequest;
        this.startTime = System.currentTimeMillis();

    }


    /**
     * 实际业务处理的响应
     *
     * @param response
     */
    public void done(RpcResponse response) {
        this.rpcResponse = rpcResponse;
        boolean release = sync.release(1); //是否释放成功
        if (release) {
            //执行多个回调函数
            invokeCallbacks();
        }

        //统计整体RPC链路调用的耗时
        long costTime = System.currentTimeMillis() - startTime;
        if (TIME_THRESHOLD < costTime) {
            log.warn("the rpc response time too slow , " +
                    "requestId+" + this.rpcRequest.getRequestId() + "cost time" + costTime);
        }


    }


    /**
     * 依次执行回调函数处理
     * 加入重入锁的原因：多个函数遍历执行
     */
    private void invokeCallbacks() {
        lock.lock();
        try {
            for (RpcCallBack callBack : pendingCallbacks) {
                runCallBack(callBack);
            }
        } finally {
            lock.unlock();
        }
    }


    /**
     * 异步处理，将response的结果，封装到future模型里
     *
     * @param callBack
     */
    private void runCallBack(final RpcCallBack callBack) {
        final RpcResponse response = this.rpcResponse;

        threadPoolExecutor.submit( ()->{
                if (response.getThrowable() == null) {
                    callBack.success(response.getResult());
                } else {
                    callBack.failer(response.getThrowable());
                }
        });

    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isDone() {
        return sync.isDone();
    }


    /**
     * 一直阻塞
     *
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */

    @Override
    public Object get() throws InterruptedException, ExecutionException {
        sync.tryAcquire(-1);    //一直阻塞: 获取许可，获取连接
        if (rpcResponse != null) {
            return rpcResponse.getResult();
        } else {
            return null;
        }

    }


    /**
     * 定时阻塞
     *
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws TimeoutException
     */
    public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        boolean success = sync.tryAcquireNanos(-1, unit.toNanos(timeout));//根据设置的定时周期，进行获取许可
        if (success) {
            if (rpcResponse != null) {
                return rpcResponse.getResult();
            } else {
                return null;
            }
        } else {
            throw new RuntimeException("timeout excuter requestId:"
                    + rpcRequest.getRequestId() +
                    "requestClassNAme:" + rpcRequest.getClassName() +
                    "requestMethondName" + rpcRequest.getMethondName());
        }
    }


    /**
     * 可以在应用处理的过程中，添加回调处理函数
     *
     * @param callBcak
     * @return
     */
    public RpcFuture addCallBack(RpcCallBack callBcak) {
        lock.lock();
        try {
            if (isDone()) {
                runCallBack(callBcak);     //判断是否执行完成，如果完成直接执行新的业务处理
            } else {
                pendingCallbacks.add(callBcak);  //添加到容器里，等待后期进行处理
            }
        } finally {
            lock.unlock();
        }
        return this;
    }


}


//实现互斥锁（自定义锁）
class Sync extends AbstractQueuedSynchronizer {

    private final int pending = 0;

    private final int done = 1;


    /**
     * 获取许可的方法
     *
     * @return
     */
    @Override
    protected boolean tryAcquire(int acquires) {
        return getState() == done ? true : false;
    }


    /**
     * 释放许可
     *
     * @param release
     * @return
     */
    @Override
    protected boolean tryRelease(int release) {
        if (getState() == pending) {

            //在pending的状态，当到达done时才进行释放许可
            if (compareAndSetState(pending, done)) {
                return true;
            }
        }
        return false;

    }


    /**
     * 是否为done的状态
     *
     * @return
     */
    public boolean isDone() {
        return getState() == done;
    }
}



